package pku;

import java.io.File;
import java.util.*;

/**
 * 消费者
 */

public class Consumer {
    List<String> topics = new LinkedList<>();
    int readPos = 0;
    String queue;
    private MessageReader[] readers;
    private int thCount = 0;
    private ByteMessage msg;

    //将消费者订阅的topic进行绑定
    public void attachQueue(String queueName, Collection<String> t) throws Exception {
        if (queue != null) {
            throw new Exception("只允许绑定一次");
        }
        queue = queueName;
        topics.addAll(t);

        try {
            TreeSet<String> topicSets = new TreeSet<>();
            for (String topic : topics) {
                String dir = Constants.DIR + File.separator + topic;
                File file = new File(dir);
                if (file.exists()) {
                    topicSets.add(dir);
                }
            }

            readers = new MessageReader[topicSets.size()];
            for (String dir : topicSets) {
                readers[thCount++] = new MessageReader(dir);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //每次消费读取一个message
    public ByteMessage poll() {
        try {
            while (readPos < thCount) {
                if ((msg = readers[readPos].read()) != null) {
                    return msg;
                } else {
                    readPos++;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;

    }

}
